//package cn.ac.iie.connect.consumer;
//
//import avro.shaded.com.google.common.collect.Lists;
//import cn.ac.iie.connect.processor.Processor;
//import com.alibaba.fastjson.JSONObject;
//import com.datastax.driver.core.*;
//import com.datastax.driver.core.exceptions.NoHostAvailableException;
//import com.datastax.driver.core.exceptions.OperationTimedOutException;
//import com.datastax.driver.core.exceptions.QueryExecutionException;
//import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
//import com.datastax.driver.core.policies.DefaultRetryPolicy;
//import com.datastax.driver.core.policies.TokenAwarePolicy;
//import org.apache.commons.codec.DecoderException;
//import org.apache.commons.codec.binary.Hex;
//import org.codehaus.jackson.annotate.JsonCreator;
//import org.codehaus.jackson.annotate.JsonProperty;
//import org.codehaus.jackson.map.DeserializationConfig;
//import org.codehaus.jackson.map.ObjectMapper;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import java.io.IOException;
//import java.nio.ByteBuffer;
//import java.util.*;
//import java.util.concurrent.atomic.AtomicBoolean;
//import java.util.concurrent.locks.LockSupport;
//import java.util.stream.Collectors;
//
///**
// * Created by jinsheng on 16/9/18.
// *
// */
//public class CassandraConsumer extends AbsConsumer {
//
//    protected String tableName;
//    private String timestampKey;
//    private Boolean isUpdate;
//    private List<String> updateFields;
//    private String updatePrimaryKey;
//
//    private List<ColumnMetadata> columns;
//    private Map<String, Type> columnMap = new HashMap<>();
//
//    protected Cluster cluster;
//    protected Session session;
//
//    private ObjectMapper objectMapper;
//    protected PreparedStatement insert_statement = null;
//    protected String exists = "";
//    protected String CQL="";
//    private static Logger LOG = LoggerFactory.getLogger(CassandraConsumer.class);
//    private static final AtomicBoolean stop = new AtomicBoolean(false);
//
//    @SuppressWarnings("unused")
//    @JsonCreator
//    public CassandraConsumer(@JsonProperty("Keyspace") String keyspace,
//                             @JsonProperty("DataCenter") String dataCenter,
//                             @JsonProperty("CoordinatorAddress") String coordinatorAddress,
//                             @JsonProperty("UserName") String userName,
//                             @JsonProperty("Password") String password,
//                             @JsonProperty("TableName") String tableName,
//                             @JsonProperty("TimestampKey") String timestampKey,
//                             @JsonProperty("isUpdate") Boolean isUpdate,
//                             @JsonProperty("updateFields") List<String> updateFields,
//                             @JsonProperty("updatePrimaryKey") String updatePrimaryKey,
//                             @JsonProperty("CQL") String CQL,
//                             @JsonProperty("TimeOut") Integer readTimeOut,
//                             @JsonProperty("EnableRetry") Boolean retry) {
//        this.tableName = tableName;
//        this.timestampKey = timestampKey;
//        this.isUpdate = isUpdate;
//        this.updateFields = updateFields;
//        this.CQL = CQL;
//        this.updatePrimaryKey = updatePrimaryKey;
//        if(readTimeOut == null) {
//            readTimeOut = 12000;
//        }
//
//        //创建连接
//        QueryOptions options = new QueryOptions();
//        options.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
//        Cluster.Builder builder = Cluster.builder();
//        if (userName.length() > 0) //需要用户认证
//            builder
//                    .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
//                    .withLoadBalancingPolicy(new TokenAwarePolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build()))
//                    //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
//                    .withCredentials(userName, password).addContactPoint(coordinatorAddress).withQueryOptions(options);
//        else // 不需要用户认证
//            builder
//                    .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
//                    .withLoadBalancingPolicy(new TokenAwarePolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build()))
//                    //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
//                    .addContactPoint(coordinatorAddress).withQueryOptions(options);
//        //设置重试策略
//        if(retry == null || retry)
//            this.cluster = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE).build();
//        else
//            this.cluster = builder.build();
//        //绑定keyspace
//        this.session = this.cluster.connect(keyspace);
//        //获取该表的元数据
//        this.columns = this.cluster.getMetadata().getKeyspace(keyspace).getTable(this.tableName).getColumns();
//
//        //处理每一列的类型并保存，用于更新时确定调用哪个set方法
//        columnMap = columns.parallelStream()
//                .collect(Collectors.toMap(ColumnMetadata::getName, col-> resolveType(col.getType())));
//        LOG.info("get column type: {}", columnMap);
//        objectMapper = new ObjectMapper();
//        objectMapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
//
//    }
//
//    /**
//     * 映射Cassandra类型到{@linkplain Type}枚举
//     * @param rawType Cassandra类型对象
//     * @return 返回Cassandra类型对应的Type枚举
//     */
//    private Type resolveType(DataType rawType) {
//        switch (rawType.getName()) {
//            case ASCII:
//            case VARCHAR:
//            case TEXT:
//                return Type.STRING;
//            case TINYINT:
//                return Type.TINYINT;
//            case SMALLINT:
//                return Type.SMALLINT;
//            case INT:
//                return Type.INT;
//            case BIGINT:
//            case VARINT:
//                return Type.BIGINT;
//            case FLOAT:
//                return Type.FLOAT;
//            case DOUBLE:
//                return Type.DOUBLE;
//            case BLOB:
//                return Type.BLOB;
//            case BOOLEAN:
//                return Type.BOOLEAN;
//            case LIST:
//                return Type.LIST;
//            case MAP:
//                return Type.MAP;
//            case SET:
//                return Type.SET;
//
//            case COUNTER:  //cuiying
//                return Type.COUNTER;
//            default:
//                throw new IllegalArgumentException("Unsupported type: "+rawType);
//        }
//    }
//
//
//    public Session getSession() {
//        return this.session;
//    }
//
//    /**
//     * 设置更新条件
//     * @return return update cql string
//     */
//    private String getUpdateStr() {
//        StringBuilder sb = new StringBuilder();
//        // 遍历需要更新的各个字段并写入where条件中
///*
//        for (String updateField : updateFields) {
//            sb.append(updateField)
//                    .append("=:")
//                    .append(updateField)
//                    .append(",");
//        }
//*///good
//
//        for (String updateField : updateFields) {
//            sb.append(updateField)
//                    .append("=")
//                    .append(updateField)
//                    .append("+")
//                    .append("0")
//                    .append(",");
//        }
//
//        sb.deleteCharAt(sb.length() - 1);
//        return sb.toString();
//    }
//
//    /**
//     * 设置更新列的值
//     * @param map 从消息队列中读取的数据
//     * @param bs bound statement
//     */
//    private void setUpdateValue(Map map, BoundStatement bs) {
//        // 遍历需要更新的各个列并分别设置更新值
//        for (String updateField : updateFields) {
//            Type type = columnMap.get(updateField);
//            Object val = map.get(updateField);
//            type.bind(updateField, val, bs);
//        }
//    }
//
//
//    /**
//     * 发送数据，传入参数是 List&lt;Map&gt; 类型的
//     * @param dataLists data list
//     */
//    @Override
//    public int pushData(List<Map<String, Object>> dataLists) throws IOException {
//        LOG.info(dataLists.toString());
//        return push_multi_json_data(dataLists);
//    }
//
//    @Override
//    protected void pushData(Map<String, Object> data) throws IOException {
//
//        while(pushSingleData(data)) {
//            LockSupport.parkNanos(1000000000);
//        }
//    }
//
//    private boolean pushSingleData(Map<String, Object> data) throws IOException {
//        if(isUpdate) {
//
//            return updateSingleData_2(data);
//        } else {
//
//            return insertSingleData(data);
//        }
//    }
//
// /*
//    private boolean pushSingleData(Map<String, Object> data) throws IOException {
//        LOG.info("777777777777777");
//        if(!insertSingleData(data)){
//            LOG.info("aaaaaaaaaaa");
//            return false;
//        }else{
//            LOG.info("bbbbbbbbbbb");
//            return updateSingleData(data);
//        }
//    }
//    */
//    /**
//     * 提交数据到Cassandra
//     *
//     * @param data 数据列表
//     * @return 返回是否提交成功
//     * @throws IOException 抛出异常
//     */
//    private boolean insertSingleData(Map<String, Object> data) throws IOException {
//
//        if(insert_statement == null) {
//            // 预编译insert语句
//            String sql = "INSERT INTO " + this.tableName + " JSON :json  if not exists";
//            //prepare(sql);
//            LOG.info(sql);
//            insert_statement = session.prepare(sql);
//        }
//
//        BoundStatement bs;
//
//        // 绑定数据并创建BoundStatement对象
//        String json = convertJson(data);
//        bs = insert_statement.bind(json);
//        LOG.debug(json);
//        return executeQuery(data.get(updatePrimaryKey), bs);
//    }
//
//    protected String convertJson(Map<String, Object> data) throws IOException {
//        return objectMapper.writeValueAsString(data);
//    }
//
//    private synchronized void prepare(String prepareCql) {
//        if(insert_statement == null) {
//            insert_statement = session.prepare(prepareCql);
//        }
//    }
//
//    /**
//     * 提交数据到Cassandra
//     * @param jsonList 数据列表
//     * @return 返回是否提交成功
//     * @throws IOException 抛出异常
//     */
//    private int push_multi_json_data(List<Map<String, Object>> jsonList) throws IOException {
//        int wrote = 0;
//        for (Map<String, Object> data : jsonList) {
//            try {
//                pushData(data);
//                wrote++;
//            } catch (Exception e) {
//                LOG.error(e.getMessage(), e);
//            }
//        }
//        return wrote;
//    }
//    private boolean updateSingleData(Map<String, Object> data) {
//
//       // LOG.info("update data: {}", data);
//        if(insert_statement == null) {
//
//            String cql = "update " + this.tableName + " set "+getUpdateStr() + " where "
//                    + getUpdateStr() + " where " + "sa=:sa and da=:da and st=:st ";
//            LOG.info("generate prepared sql: {}", cql);
//            prepare(cql);
//        }
//        String str = (String) data.get("json");
//        JSONObject jo  = JSONObject.parseObject(str);
//        Map map = (Map) jo;
//        LOG.info(map.toString());
//        //绑定更新数据，并创建BoundStatement对象
//        BoundStatement bs = insert_statement.bind();
//        setUpdateValue(data, bs);
//        // 绑定更新记录的主键
//        columnMap.get(updatePrimaryKey).bind("sa", data.get(updatePrimaryKey), bs);
//        // 设置记录时间戳
//        if (this.timestampKey.length() > 0 && data.containsKey(this.timestampKey))
//            bs.setDefaultTimestamp((Long) data.get(this.timestampKey) * 1000000); // second to microsecond
//
//        return executeQuery(data.get(updatePrimaryKey), bs);
//    }
//
//    private boolean updateSingleData_2(Map<String, Object> data) {
//
//        /*
//        String cql = "update " + this.tableName + " set "+getUpdateStr() + " where "
//                + getUpdateStr() + " where " + "sa=:sa and da=:da and st=:st if exists";
//        LOG.info("generate prepared sql: {}", cql);
//        */
//        if(insert_statement == null) {
//
//            // 预编译update语句
//            /*
//            String cql = "update " + this.tableName + " set "
//                    + getUpdateStr() + " where " + this.updatePrimaryKey
//                    + "=:key "+exists;// IF EXISTS";
//            */
//
//         //   String cql = "update " + this.tableName + " set "+getUpdateStr() + " where " + " sa=:sa and da=:da and st=:st ";
//            String cql = this.CQL;
//            //  String cql = " updatae "+ this.tableName + " set fib=fib+0,fob=fob+0 where sa='123.128.14.178' and da='114.255.165.2' and st=1505440800000000" ;
//            LOG.info("generate prepared sql: {}", cql);
//
//            prepare(cql);
//        }
//
//   //     LOG.info("update data: {}", data);
//
//        String str = (String) data.get("json");
//        JSONObject jo  = JSONObject.parseObject(str);
//        Map<String,Object> map = new HashMap<String,Object>();
//
//        map.put("sa",jo.get("sa"));
//        map.put("da",jo.get("da"));
//        map.put("st",jo.get("st"));
//        map.put("fib",jo.get("fib"));
//        map.put("fob",jo.get("fob"));
//
//        /*
//        SimpleDateFormat format =  new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//        Date date = null;
//
//        try {
//            date = format.parse("2017-09-01 00:00:00");
//            map.put("st",date.getTime()*1000);
//        } catch (ParseException e) {
//            e.printStackTrace();
//        }
//        map.put("sa","123.128.14.178");
//        map.put("da","114.255.165.2");
//        map.put("fib",17);
//        map.put("fob",19);
//        */
//      //  LOG.info(map.toString());
//        //绑定更新数据，并创建BoundStatement对象
//        BoundStatement bs = insert_statement.bind();
//      //  setUpdateValue(data, bs);
//        setUpdateValue(map , bs);
//        // 绑定更新记录的主键
//
//        columnMap.get(updatePrimaryKey).bind("sa", map.get(updatePrimaryKey), bs);
//        columnMap.get("da").bind("da",map.get("da"),bs);
//        columnMap.get("st").bind("st",map.get("st"),bs);
//        // 设置记录时间戳
//
//        if (this.timestampKey.length() > 0 && map.containsKey(this.timestampKey))
//            bs.setDefaultTimestamp((Long) map.get(this.timestampKey) * 1000000); // second to microsecond
//
//        return executeQuery(map.get(updatePrimaryKey), bs);
//    }
//
//    private boolean executeQuery(Object key, Statement st) {
//
//        try {
//            ResultSet rs = session.execute(st);
//          //  ResultSet rs = (ResultSet) session.executeAsync(st);
//            String result_string = rs.toString();
//            if (result_string.contains("ERROR")) {
//                LOG.info(result_string);
//            }
//            //LOG.warn("executed query for data: {}", key);
//        } catch (OperationTimedOutException |QueryExecutionException |NoHostAvailableException e) {
//            LOG.warn("execute failed ,sa: {}, exp: {}", key, e.getMessage());
//            return true;
//        }
//
//
//        return false;
//    }
//
//    @Override
//    public void internalClose() {
//        stop.set(true);
//        this.session.close();
//        this.cluster.close();
//    }
//
//    public List<ColumnMetadata> getColumns() {
//        return columns;
//    }
//
//    private enum Type {
//        STRING {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else
//                    bs.setString(col, val.toString());
//            }
//        }, TINYINT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof String) {
//                    bs.setByte(col, Long.valueOf(((String) val).toLowerCase()).byteValue());
//                } else if(val instanceof Number) {
//                    bs.setByte(col, ((Number)val).byteValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        }, SMALLINT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof String) {
//                    bs.setShort(col, Long.valueOf(((String) val).toLowerCase()).shortValue());
//                } else if(val instanceof Number) {
//                    bs.setShort(col, ((Number)val).shortValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        }, INT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof String) {
//                    bs.setInt(col, Long.valueOf(((String) val).toLowerCase()).intValue());
//                } else if(val instanceof Number) {
//                    bs.setInt(col, ((Number)val).intValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        }, BIGINT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof String) {
//                    bs.setLong(col, Long.valueOf(((String) val).toLowerCase()));
//                } else if(val instanceof Number) {
//                    bs.setLong(col, ((Number)val).longValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        }, FLOAT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof String) {
//                    bs.setFloat(col, Long.valueOf(((String) val).toLowerCase()).floatValue());
//                } else if(val instanceof Number) {
//                    bs.setFloat(col, ((Number)val).floatValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        }, DOUBLE {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof String) {
//                    bs.setDouble(col, Long.valueOf(((String) val).toLowerCase()).doubleValue());
//                } else if(val instanceof Number) {
//                    bs.setDouble(col, ((Number)val).doubleValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        }, BLOB {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof String) {
//                    try {
//                        byte[] bytes = Hex.decodeHex(((String)val).toCharArray());
//                        bs.setBytes(col, ByteBuffer.wrap(bytes));
//                    } catch (DecoderException e) {
//                        throw new IllegalArgumentException("wrong value of hex string, column: "+ col);
//                    }
//                } else if(val instanceof byte[]) {
//                    bs.setBytes(col, ByteBuffer.wrap((byte[]) val));
//                } else if(val instanceof ByteBuffer) {
//                    bs.setBytes(col, (ByteBuffer) val);
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        }, BOOLEAN {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof String) {
//                    bs.setBool(col, Boolean.valueOf((String) val));
//                } else if(val instanceof Boolean) {
//                    bs.setBool(col, (Boolean) val);
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        }, LIST {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof List) {
//                    bs.setList(col, (List<?>) val);
//                } else if(val instanceof Collection) {
//                    List<?> l = new ArrayList<>((Collection<?>)val);
//                    bs.setList(col, l);
//                } else {
//                    List<Object> l = new ArrayList<>();
//                    l.add(val);
//                    bs.setList(col, l);
//                }
//            }
//        }, MAP {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null) {
//                    bs.setToNull(col);
//                } else if(val instanceof Map) {
//                    bs.setMap(col, (Map<?, ?>) val);
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        }, SET {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if(val == null) {
//                    bs.setToNull(col);
//                } else if(val instanceof Set){
//                    bs.setSet(col, (Set<?>) val);
//                } else if( val instanceof Collection) {
//                    Set<?> s = new HashSet<>((Collection<?>) val);
//                    bs.setSet(col, s);
//                } else {
//                    Set<Object> s = new HashSet<>();
//                    s.add(val);
//                    bs.setSet(col, s);
//                }
//            }
//        },COUNTER{//cuiying
//            public void bind(String col,Object val , BoundStatement bs){
//                if(val == null)
//                    bs.setToNull(col);
//                else if(val instanceof String) {
//                    bs.setLong(col, Long.valueOf(((String) val).toLowerCase()));
//                } else if(val instanceof Number) {
//                    bs.setLong(col, ((Number)val).longValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of "+ col);
//                }
//            }
//        };
//
//        public void bind(String col, Object val, BoundStatement bs) {
//            throw new IllegalArgumentException("wrong type of "+ col);
//        }
//    }
//
//    @JsonProperty("exists")
//    public String getExists() {
//        return exists;
//    }
//
//    public void setExists(String exists) {
//        this.exists = exists;
//    }
//
//
//}
